package com.dahuan.tables.udf;

import com.dahuan.bean.SensorReading;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.apache.flink.table.functions.AggregateFunction;
import org.apache.flink.types.Row;

public class UDF_AggregateFunction {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism( 1 );
        //指定事件时间
        env.setStreamTimeCharacteristic( TimeCharacteristic.EventTime );

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create( env );

        // 2. 读入文件数据，得到DataStream
        DataStream<String> inputStream = env.readTextFile( "E:\\Project\\FlinkTutorials\\Flink-Scala\\src\\main\\resources\\sensor.txt" );

        // 3. 转换成POJO
        DataStream<SensorReading> dataStream = inputStream.map( line -> {
            String[] fields = line.split( "," );
            return new SensorReading( fields[0], new Long( fields[1] ), new Double( fields[2] ) );
        } );

        // 4. 将流转换成表，定义时间特性
        Table sensorTable = tableEnv.fromDataStream( dataStream, "id, timestamp as ts, temperature as temp" );

        // 5. 自定义聚合函数 求当前传感器的平均温度值
        AvgTemp avgTemp = new AvgTemp();

        // 需要在环境中注册UDF
        tableEnv.registerFunction("avgTemp", avgTemp);
        Table resultTable = sensorTable
                .groupBy("id")
                .aggregate( "avgTemp(temp) as avgtemp" )
                .select("id, avgtemp");

        // 4.2 SQL
        tableEnv.createTemporaryView("sensor", sensorTable);
        Table resultSqlTable = tableEnv.sqlQuery("select id, avgTemp(temp) " +
                " from sensor group by id");

        // 打印输出
        tableEnv.toRetractStream(resultTable, Row.class).print("result");
        tableEnv.toRetractStream(resultSqlTable, Row.class).print("sql");



        env.execute("UDF_AggregateFunction");
    }

    //实现自定义的AggregateFunction
    public static class AvgTemp extends AggregateFunction<Double, Tuple2<Double,Integer>>{

        @Override
        public Double getValue(Tuple2<Double, Integer> accumulator) {
            return accumulator.f0 / accumulator.f1;
        }

        @Override
        public Tuple2<Double, Integer> createAccumulator() {
            return new Tuple2<>(0.0,0);
        }


        //必须实现一个accumulate方法，来数据之后更新状态
        public void accumulate(Tuple2<Double,Integer> accumulator,Double temp ){

            accumulator.f0 += temp;
            accumulator.f1 += 1;

        }

    }
}
